'use strict';

const { Kafka, logLevel } = require('kafkajs');

class KafkaModel {
    constructor(brokers) {
        this.KafkaInstance = new Kafka({
            clientId: 'my-app',
            brokers,
            retry: {
                retries: 8
            },
            logLevel: logLevel.ERROR
        });
    }
    async producer(topic, groupId, msg) {
        try {
            const producer = this.KafkaInstance.producer({ groupId });
            await producer.connect();
            await producer.send({
                topic,
                messages: [
                    {
                        value: JSON.stringify(msg)
                    }
                ]
            });
        } catch (error) {
            throw error;
        }
    }

    async consumer(topic, groupId, callback) {
        const consumer = this.KafkaInstance.consumer({ groupId });
        await consumer.connect();
        await consumer.subscribe({ topic });
        await consumer.run({
            eachMessage: async ({ message }) => {
                await callback(JSON.parse(message.value.toString()));
            }
        });
    }
}
module.exports = KafkaModel;
